package com.mng.tableAPI;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.sinks.CsvTableSink;
import static org.apache.flink.table.api.Expressions.$;

public class TableCreatTest {
    public static void main(String[] args) throws Exception{
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);


//        tableEnv.connect(new FileSystem().path("src/main/resources/wc.txt"))
//                .withFormat(new Csv())
//                .withSchema(new Schema()
//                        .field("id", DataTypes.STRING())
//                        .field("timestamp", DataTypes.BIGINT())
//                        .field("temperaturre", DataTypes.DOUBLE())
//                )
//                .createTemporaryTable("inputTable");
//        Table inputTable = tableEnv.from("inputTable");
//
//        inputTable.printSchema();
//
//        tableEnv.toAppendStream(inputTable, Row.class).print();


        tableEnv.connect(new Kafka()
                    .version("universal")//Flink附带了提供了多个Kafka连接器：universal通用版本，0.10，0.11
                    .topic("sensor")
                    .startFromEarliest()
//                    .property("zookeeper.connect", "47.94.161.48:2181")
                    .property("bootstrap.servers", "47.94.161.48:9092"))
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temperaturre", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");
        Table inputTable = tableEnv.from("inputTable");
        Table sensor1Table = inputTable.select($("id"), $("temperaturre"))
                .where($("id").isEqual("sensor_1"));

        tableEnv.connect(
                new Kafka()
                        .version("universal")//Flink附带了提供了多个Kafka连接器：universal通用版本，0.10，0.11
                        .topic("sinktest")
                        .startFromEarliest()
//                    .property("zookeeper.connect", "47.94.161.48:2181")
                        .property("bootstrap.servers", "47.94.161.48:9092")
        )
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id",DataTypes.STRING())
//                        .field("timestamp",DataTypes.BIGINT())
                        .field("temperaturre",DataTypes.DOUBLE())
                )
                .createTemporaryTable("outputTable");

        sensor1Table.insertInto("outputTable");


        //env.execute();
       tableEnv.execute("");

    }
}